#include "server.h"

struct
{/*{{{*/
    unsigned long thrd_id;
    EP ep;
    EP_SLOT listener_slot;
    THRD_MSGQ *worker_msgqs;
}/*}}}*/
listener_ctl;

static int _conn(struct ep_active_slot *active_slot)
{/*{{{*/
#define _MAX_CONN_NUM        100
    SOCKET new_sock=active_slot->_u._conn.sock;

    pthread_mutex_lock(&stat_info._worker_.lock);

    int idle_num=0;
    int idle[cnfg_info.workers];
    int min_busy_conn_num=_MAX_CONN_NUM;
    int min_busy_idx=-1;
    int worker_idx;
    for(worker_idx=0; worker_idx<cnfg_info.workers; ++worker_idx)
    {
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        if(worker->conn_num == 0)
        {   idle[idle_num++]=worker_idx;   }
        else
        {
            if(min_busy_conn_num > worker->conn_num)
            {
                min_busy_conn_num=worker->conn_num;
                min_busy_idx=worker_idx;
            }
        }
    }
    if(idle_num > 0) // 选取无连接的worker
    {
        worker_idx=idle[random_int(0, idle_num-1, NULL)];
#ifdef _MY_DEBUG
        errlog("%s: choose idle thread(%d)",
                __func__, worker_idx);
#endif
    }
    else if(min_busy_conn_num < _MAX_CONN_NUM) // 选取连接最少的worker
    {
        worker_idx=min_busy_idx;
    }
    else
    {
        errlog("%s: worker full, BAD", __func__);
        pthread_mutex_unlock(&stat_info._worker_.lock);
        sock_close(new_sock);
        return(-1);
    }
    ++stat_info._worker_.worker[worker_idx].conn_num;
#ifdef _MY_DEBUG
    errlog("%s: worker(%d) conn_num:%d", __func__, worker_idx,
            stat_info._worker_.worker[worker_idx].conn_num);
#endif

    pthread_mutex_unlock(&stat_info._worker_.lock);

    struct msgnode *node;
    ALLOC_DOERR(node, struct msgnode,
            errlog("%s: listener alloc msgnode fail", __func__);
            sock_close(new_sock);
            );
    strcpy(node->svc, "NEWCONN");
    node->access_ts=curr_timestamp(NULL);
    CLR_BUF(node->msg);
    para_put_item_int(node->msg, new_sock);
    thrd_msgq_send(listener_ctl.worker_msgqs[worker_idx], &node->link);

    return(0);
}/*}}}*/

static void _listener()
{/*{{{*/
#define CHECK_OPERATE_INFO_INTERVAL         1
    unsigned int quit=IS_FALSE;
    struct ep_active_slot *active_list;
    time_t last_operate=0;
    time_t now;
    int res;
    
    while(!quit)
    {
        res=ep_poll(listener_ctl.ep, &active_list);
        if(res)
        {
            errlog("%s: ep_poll fail, res:%d, impossible",
                    __func__, res);
            continue;
        }
        while(active_list)
        {/*{{{*/
            switch(active_list->type)
            {
                case EP_SLOT_CONN:
                    _conn(active_list);
                    break;
                default:
                    errlog("%s: listener poll type(%s) invalid, "
                            "impossible",
                            __func__, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }/*}}}*/

        now=curr_timestamp(NULL);

        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {/*{{{*/
            pthread_mutex_lock(&operate_info._listener_.lock); //lock
            quit=operate_info._listener_.quit;
            if(quit)
            {
                operate_info._listener_.state=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._listener_.lock); //unlock

            last_operate=now;
        }/*}}}*/
    }
    DEEP_SLEEP;
}/*}}}*/

int listener_start(THRD_MSGQ *msgqs)
{/*{{{*/
    char buf[200];
    int res;

    // operate_info
    pthread_mutex_init(&operate_info._listener_.lock, NULL);
    operate_info._listener_.quit=IS_FALSE;
    operate_info._listener_.state=THRD_RUNNING;

    res=ep_create(&listener_ctl.ep);
    if(res)
    {
        sprintf(buf, "%s: listener(ep_create) fail, res:%d",
                __func__, res);
        fprintf(stderr, "%s:\n", buf); errlog(buf);
        return(-1);
    }

    // server port
    SOCKET sock=sock_listen(cnfg_info._access_.server_port);
    if(sock == INVALID_SOCKET)
    {
        sprintf("%s: open server port fail", __func__);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    res=ep_slot_register(listener_ctl.ep, sock, IS_TRUE,
            &listener_ctl.listener_slot);
    if(res)
    {
        sprintf(buf,
                "%s: listener(ep_slot_register) server port fail, res:%d",
                __func__, res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    sprintf(buf, "%s: open server port(%d)",
            __func__, cnfg_info._access_.server_port);
    fprintf(stderr, "%s\n", buf); errlog(buf);

    // listener-worker channel
    listener_ctl.worker_msgqs=msgqs;

    // start
    res=thrd_create((THRD_FUNC_T)_listener, NULL, THRD_DETACH,
                &listener_ctl.thrd_id);
    if(res)
    {
        sprintf(buf, "%s: listener thrd create fail, res:%d",
                __func__, res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }

    sprintf(buf, "%s: workers thrd create succeed", __func__);
    fprintf(stderr, "%s\n", buf); errlog(buf);
    return(0);
}/*}}}*/
